Micron Document
πŸŽ–οΈGitΠ―Ρ€Π°πŸŽ–οΈ

Node / meshtastic / Meshtastic-Android / files / core / network / src / commonMain / kotlin / org / meshtastic / core / network / transport / StreamFrameCodec.kt

Displaying Raw β€’ Download

core/network/src/commonMain/kotlin/org/meshtastic/core/network/transport/StreamFrameCodec.kt b4bedd92fcc8d65daff79200d71f6ebc3d8f0cfa (b4bedd92) Text, 9.50 KB

T8b949e/*
* Copyright (c) 2026 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
Tff7b72package T7ee787org.meshtastic.core.network.transport

Tff7b72import T7ee787co.touchlab.kermit.Logger
Tff7b72import T7ee787kotlinx.coroutines.sync.Mutex
Tff7b72import T7ee787kotlinx.coroutines.sync.withLock
Tff7b72import T7ee787kotlin.concurrent.Volatile

T8b949e/**
* Meshtastic stream framing codec β€” pure Kotlin, no platform dependencies.
*
* Implements the START1/START2 + 2-byte-length + payload framing protocol used for serial and TCP communication with
* Meshtastic radios.
*
* Shared across Android, Desktop, and iOS via `SharedRadioInterfaceService`.
*/
Tf0883e@SuppressTb4b4b4(Ta5d6ff"Ta5d6ffMagicNumberTa5d6ff"Tb4b4b4)
Tff7b72class T56d364StreamFrameCodecTb4b4b4(
T8b949e/** Called when a complete packet has been decoded from the byte stream. */
Tff7b72private Tff7b72val Te6edf3onPacketReceivedTb4b4b4: Tb4b4b4(Te6edf3ByteArrayTb4b4b4) Tff7b72-Tff7b72> Tffa657UnitTb4b4b4,
T8b949e/** Optional log tag for debug output. */
Tff7b72private Tff7b72val Te6edf3logTagTb4b4b4: Tffa657String Tff7b72= Ta5d6ff"Ta5d6ffStreamCodecTa5d6ff"Tb4b4b4,
Tb4b4b4) Tb4b4b4{
Tff7b72companion Tff7b72object Tb4b4b4{
Tff7b72const Tff7b72val Te6edf3START1Tb4b4b4: Tffa657Byte Tff7b72= T79c0ff0Te6edf3x94Tb4b4b4.Te6edf3toByteTb4b4b4(Tb4b4b4)
Tff7b72const Tff7b72val Te6edf3START2Tb4b4b4: Tffa657Byte Tff7b72= T79c0ff0Te6edf3xc3Tb4b4b4.Te6edf3toByteTb4b4b4(Tb4b4b4)
Tff7b72const Tff7b72val Te6edf3MAX_TO_FROM_RADIO_SIZE Tff7b72= T79c0ff5T79c0ff1T79c0ff2
Tff7b72const Tff7b72val Te6edf3HEADER_SIZE Tff7b72= T79c0ff4

T8b949e/** Cap on the device-log line buffer. Bounds [debugOut] for a stream that never yields a newline. */
Tff7b72const Tff7b72val Te6edf3MAX_DEBUG_LINE_LENGTH Tff7b72= T79c0ff5T79c0ff1T79c0ff2

T8b949e/** Default Meshtastic TCP service port. */
Tff7b72const Tff7b72val Te6edf3DEFAULT_TCP_PORT Tff7b72= T79c0ff4T79c0ff4T79c0ff0T79c0ff3

T8b949e/** Wake bytes to send before connecting to rouse a sleeping device. */
Tff7b72val Te6edf3WAKE_BYTES Tff7b72= Te6edf3byteArrayOfTb4b4b4(Te6edf3START1Tb4b4b4, Te6edf3START1Tb4b4b4, Te6edf3START1Tb4b4b4, Te6edf3START1Tb4b4b4)
Tb4b4b4}

Tff7b72private Tff7b72val Te6edf3writeMutex Tff7b72= Te6edf3MutexTb4b4b4(Tb4b4b4)

T8b949e// Framing state machine
Tff7b72private Tff7b72var Te6edf3ptr Tff7b72= T79c0ff0
Tff7b72private Tff7b72var Te6edf3msb Tff7b72= T79c0ff0
Tff7b72private Tff7b72var Te6edf3lsb Tff7b72= T79c0ff0
Tff7b72private Tff7b72var Te6edf3packetLen Tff7b72= T79c0ff0
Tff7b72private Tff7b72val Te6edf3rxPacket Tff7b72= Te6edf3ByteArrayTb4b4b4(Te6edf3MAX_TO_FROM_RADIO_SIZETb4b4b4)
Tff7b72private Tff7b72val Te6edf3debugLineBuf Tff7b72= Te6edf3StringBuilderTb4b4b4(Tb4b4b4)

T8b949e// Written only by the transport's read thread, but published through [framingDesyncCount] for diagnostics, so a
T8b949e// reader on another thread needs a visibility guarantee. Matches the metrics counters in TcpTransport. The rest of
T8b949e// the state machine stays plain: it is single-reader by contract.
Tf0883e@Volatile Tff7b72private Tff7b72var Te6edf3desyncCount Tff7b72= T79c0ff0L

T8b949e// Purely local to the read thread β€” it only throttles logging β€” so it needs no cross-thread guarantee.
Tff7b72private Tff7b72var Te6edf3consecutiveDesyncs Tff7b72= T79c0ff0

T8b949e/**
* Number of framing desyncs observed since the last [reset].
*
* Exposed for diagnostics and tests. A non-zero count on an otherwise healthy link means the peer is emitting bytes
* we cannot frame β€” truncated writes, a corrupt length prefix, or non-protocol output on the wire.
*/
Tff7b72val Te6edf3framingDesyncCountTb4b4b4: Tffa657Long
Tff7b72getTb4b4b4(Tb4b4b4) Tff7b72= Te6edf3desyncCount

T8b949e/**
* Process a single incoming byte through the stream framing state machine.
*
* Call this repeatedly with bytes from the transport (serial, TCP, etc). When a complete packet is decoded,
* [onPacketReceived] is invoked.
*
* Malformed input never throws. On a broken frame the machine records a desync, rewinds to the hunting state, and
* resynchronizes on the next valid frame header β€” a corrupt frame costs one frame, not the connection.
*/
Tff7b72fun Td2a8ffprocessInputByteTb4b4b4(Te6edf3cTb4b4b4: Tffa657ByteTb4b4b4) Tb4b4b4{
T8b949e// A byte that breaks framing may itself be the START1 of the next frame, so re-examine it from the hunting
T8b949e// state instead of dropping it. This matters most for a corrupt length prefix whose low byte is 0x94: the
T8b949e// machine rejects the length and the very next header would otherwise be missing its start byte.
T8b949e//
T8b949e// Discarding the offending byte is what let one bad byte cascade. The machine would resume hunting partway
T8b949e// into the *payload* of the frame that followed, re-sync on an incidental 0x94/0xc3 pair inside it, and hand
T8b949e// a misaligned byte range to the protobuf parser β€” surfacing downstream as ProtocolException or
T8b949e// "Unexpected call to beginMessage()".
T8b949e//
T8b949e// The retry always runs in state 0, which never reports a desync, so this terminates after at most one retry.
Tff7b72if Tb4b4b4(Tff7b72!Te6edf3stepTb4b4b4(Te6edf3cTb4b4b4)Tb4b4b4) Te6edf3stepTb4b4b4(Te6edf3cTb4b4b4)
Tb4b4b4}

T8b949e/**
* Advance the state machine by a single byte.
*
* @return `false` when [c] broke framing and must be re-examined from the hunting state.
*/
Tff7b72private Tff7b72fun Td2a8ffstepTb4b4b4(Te6edf3cTb4b4b4: Tffa657ByteTb4b4b4)Tb4b4b4: Tffa657Boolean Tff7b72= Tff7b72when Tb4b4b4(Te6edf3ptrTb4b4b4) Tb4b4b4{
T79c0ff0 Tff7b72-Tff7b72> Tb4b4b4{
T8b949e// Hunting for a frame start; anything else is device debug output.
Tff7b72if Tb4b4b4(Te6edf3c Tff7b72=Tff7b72= Te6edf3START1Tb4b4b4) Te6edf3ptr Tff7b72= T79c0ff1 Tff7b72else Te6edf3debugOutTb4b4b4(Te6edf3cTb4b4b4)
Tff7b72true
Tb4b4b4}

T79c0ff1 Tff7b72-Tff7b72> Te6edf3stepExpectStart2Tb4b4b4(Te6edf3cTb4b4b4)

T79c0ff2 Tff7b72-Tff7b72> Tb4b4b4{
Te6edf3msb Tff7b72= Te6edf3cTb4b4b4.Te6edf3toIntTb4b4b4(Tb4b4b4) Te6edf3and T79c0ff0Te6edf3xff
Te6edf3ptr Tff7b72= T79c0ff3
Tff7b72true
Tb4b4b4}

T79c0ff3 Tff7b72-Tff7b72> Te6edf3stepLengthTb4b4b4(Te6edf3cTb4b4b4)

Tff7b72else Tff7b72-Tff7b72> Tb4b4b4{
Te6edf3appendPayloadByteTb4b4b4(Te6edf3cTb4b4b4)
Tff7b72true
Tb4b4b4}
Tb4b4b4}

T8b949e/**
* State 1: expecting START2.
*
* A repeated START1 is padding rather than corruption β€” the wake sequence is four of them, and a peer may pad an
* idle link the same way β€” so hold in this state and take the next byte as the candidate START2.
*/
Tff7b72private Tff7b72fun Td2a8ffstepExpectStart2Tb4b4b4(Te6edf3cTb4b4b4: Tffa657ByteTb4b4b4)Tb4b4b4: Tffa657Boolean Tff7b72= Tff7b72when Tb4b4b4(Te6edf3cTb4b4b4) Tb4b4b4{
Te6edf3START2 Tff7b72-Tff7b72> Tb4b4b4{
Te6edf3ptr Tff7b72= T79c0ff2
Tff7b72true
Tb4b4b4}

Te6edf3START1 Tff7b72-Tff7b72> Tb4b4b4{
T8b949e// Padding β€” hold here, still expecting START2.
Tff7b72true
Tb4b4b4}

Tff7b72else Tff7b72-Tff7b72> Tb4b4b4{
Te6edf3lostSyncTb4b4b4(Ta5d6ff"Ta5d6ffexpected START2Ta5d6ff"Tb4b4b4)
Tff7b72false
Tb4b4b4}
Tb4b4b4}

T8b949e/** State 3: low byte of the length prefix, completing the header. */
Tff7b72private Tff7b72fun Td2a8ffstepLengthTb4b4b4(Te6edf3cTb4b4b4: Tffa657ByteTb4b4b4)Tb4b4b4: Tffa657Boolean Tb4b4b4{
Te6edf3lsb Tff7b72= Te6edf3cTb4b4b4.Te6edf3toIntTb4b4b4(Tb4b4b4) Te6edf3and T79c0ff0Te6edf3xff
Te6edf3packetLen Tff7b72= Tb4b4b4(Te6edf3msb Te6edf3shl T79c0ff8Tb4b4b4) Te6edf3or Te6edf3lsb
Tff7b72return Tff7b72when Tb4b4b4{
Te6edf3packetLen Tff7b72> Te6edf3MAX_TO_FROM_RADIO_SIZE Tff7b72-Tff7b72> Tb4b4b4{
Te6edf3lostSyncTb4b4b4(Ta5d6ff"Ta5d6ffdeclared length Tffd700$Te6edf3packetLenTa5d6ff exceeds Tffd700$Te6edf3MAX_TO_FROM_RADIO_SIZETa5d6ff"Tb4b4b4)
Tff7b72false
Tb4b4b4}

Te6edf3packetLen Tff7b72=Tff7b72= T79c0ff0 Tff7b72-Tff7b72> Tb4b4b4{
Te6edf3deliverPacketTb4b4b4(Tb4b4b4)
Tff7b72true
Tb4b4b4}

Tff7b72else Tff7b72-Tff7b72> Tb4b4b4{
Te6edf3ptr Tff7b72= Te6edf3HEADER_SIZE
Tff7b72true
Tb4b4b4}
Tb4b4b4}
Tb4b4b4}

T8b949e/** States 4 and up: accumulating the payload declared by the header. */
Tff7b72private Tff7b72fun Td2a8ffappendPayloadByteTb4b4b4(Te6edf3cTb4b4b4: Tffa657ByteTb4b4b4) Tb4b4b4{
Te6edf3rxPacketTff7b72[Te6edf3ptr Tff7b72- Te6edf3HEADER_SIZETff7b72] Tff7b72= Te6edf3c
Tff7b72if Tb4b4b4(Te6edf3ptr Tff7b72- Te6edf3HEADER_SIZE Tff7b72+ T79c0ff1 Tff7b72=Tff7b72= Te6edf3packetLenTb4b4b4) Te6edf3deliverPacketTb4b4b4(Tb4b4b4) Tff7b72else Te6edf3ptrTff7b72+Tff7b72+
Tb4b4b4}

Tff7b72private Tff7b72fun Td2a8fflostSyncTb4b4b4(Te6edf3reasonTb4b4b4: Tffa657StringTb4b4b4) Tb4b4b4{
Te6edf3desyncCountTff7b72+Tff7b72+
Te6edf3consecutiveDesyncsTff7b72+Tff7b72+
T8b949e// Structural detail only β€” never payload bytes, which can carry message content or keys.
Tff7b72val Te6edf3detail Tff7b72= Ta5d6ff"Tffd700$Te6edf3logTagTa5d6ff: Lost protocol sync (Tffd700$Te6edf3reasonTa5d6ff); resynchronizing. Desyncs since connect: Tffd700$Te6edf3desyncCountTa5d6ff"
T8b949e// Warn once per disruption, then drop to debug: a peer streaming non-protocol bytes would otherwise emit a
T8b949e// warning every few hundred bytes.
Tff7b72if Tb4b4b4(Te6edf3consecutiveDesyncs Tff7b72=Tff7b72= T79c0ff1Tb4b4b4) Te6edf3LoggerTb4b4b4.Te6edf3w Tb4b4b4{ Te6edf3detail Tb4b4b4} Tff7b72else Te6edf3LoggerTb4b4b4.Te6edf3d Tb4b4b4{ Te6edf3detail Tb4b4b4}
Te6edf3ptr Tff7b72= T79c0ff0
Tb4b4b4}

Tff7b72private Tff7b72fun Td2a8ffdeliverPacketTb4b4b4(Tb4b4b4) Tb4b4b4{
Te6edf3consecutiveDesyncs Tff7b72= T79c0ff0
T8b949e// Rewind before dispatching so the machine is already hunting if the callback re-enters.
Te6edf3ptr Tff7b72= T79c0ff0
Te6edf3onPacketReceivedTb4b4b4(Te6edf3rxPacketTb4b4b4.Te6edf3copyOfTb4b4b4(Te6edf3packetLenTb4b4b4)Tb4b4b4)
Tb4b4b4}

T8b949e/**
* Frames a payload into the Meshtastic stream protocol format: [START1][START2][MSB len][LSB len][payload].
*
* Thread-safe via an internal mutex β€” multiple callers can call this concurrently.
*/
Tff7b72suspend Tff7b72fun Td2a8ffframeAndSendTb4b4b4(
Te6edf3payloadTb4b4b4: Te6edf3ByteArrayTb4b4b4,
Te6edf3sendBytesTb4b4b4: Te6edf3suspend Tb4b4b4(Te6edf3ByteArrayTb4b4b4) Tff7b72-Tff7b72> Tffa657UnitTb4b4b4,
Te6edf3flushTb4b4b4: Te6edf3suspend Tb4b4b4(Tb4b4b4) Tff7b72-Tff7b72> Tffa657Unit Tff7b72= Tb4b4b4{Tb4b4b4}Tb4b4b4,
Tb4b4b4) Tb4b4b4{
Te6edf3writeMutexTb4b4b4.Te6edf3withLock Tb4b4b4{
Tff7b72val Te6edf3header Tff7b72= Te6edf3ByteArrayTb4b4b4(Te6edf3HEADER_SIZETb4b4b4)
Te6edf3headerTff7b72[T79c0ff0Tff7b72] Tff7b72= Te6edf3START1
Te6edf3headerTff7b72[T79c0ff1Tff7b72] Tff7b72= Te6edf3START2
Te6edf3headerTff7b72[T79c0ff2Tff7b72] Tff7b72= Tb4b4b4(Te6edf3payloadTb4b4b4.Te6edf3size Te6edf3shr T79c0ff8Tb4b4b4)Tb4b4b4.Te6edf3toByteTb4b4b4(Tb4b4b4)
Te6edf3headerTff7b72[T79c0ff3Tff7b72] Tff7b72= Tb4b4b4(Te6edf3payloadTb4b4b4.Te6edf3size Te6edf3and T79c0ff0Te6edf3xffTb4b4b4)Tb4b4b4.Te6edf3toByteTb4b4b4(Tb4b4b4)

Te6edf3sendBytesTb4b4b4(Te6edf3headerTb4b4b4)
Te6edf3sendBytesTb4b4b4(Te6edf3payloadTb4b4b4)
Te6edf3flushTb4b4b4(Tb4b4b4)
Tb4b4b4}
Tb4b4b4}

T8b949e/** Resets the framing state machine. Call when reconnecting. */
Tff7b72fun Td2a8ffresetTb4b4b4(Tb4b4b4) Tb4b4b4{
Te6edf3ptr Tff7b72= T79c0ff0
Te6edf3msb Tff7b72= T79c0ff0
Te6edf3lsb Tff7b72= T79c0ff0
Te6edf3packetLen Tff7b72= T79c0ff0
Te6edf3desyncCount Tff7b72= T79c0ff0
Te6edf3consecutiveDesyncs Tff7b72= T79c0ff0
Te6edf3debugLineBufTb4b4b4.Te6edf3clearTb4b4b4(Tb4b4b4)
Tb4b4b4}

T8b949e/**
* Print device serial debug output to the logger.
*
* Any byte stream that never yields a newline accumulates here, so the buffer is capped: a stream of non-`START1`
* bytes with no `\n` would otherwise grow the heap without limit. At the cap the line is flushed as-is and the
* buffer reset, which keeps the output readable rather than silently dropping it.
*/
Tff7b72private Tff7b72fun Td2a8ffdebugOutTb4b4b4(Te6edf3bTb4b4b4: Tffa657ByteTb4b4b4) Tb4b4b4{
Tff7b72when Tb4b4b4(Tff7b72val Te6edf3c Tff7b72= Te6edf3bTb4b4b4.Te6edf3toIntTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3toCharTb4b4b4(Tb4b4b4)Tb4b4b4) Tb4b4b4{
Ta5d6ff'\r' Tff7b72-Tff7b72> Tb4b4b4{Tb4b4b4}

Ta5d6ff'\n' Tff7b72-Tff7b72> Tb4b4b4{
Te6edf3LoggerTb4b4b4.Te6edf3d Tb4b4b4{ Ta5d6ff"Tffd700$Te6edf3logTagTa5d6ff DeviceLog: Tffd700$Te6edf3debugLineBufTa5d6ff" Tb4b4b4}
Te6edf3debugLineBufTb4b4b4.Te6edf3clearTb4b4b4(Tb4b4b4)
Tb4b4b4}

Tff7b72else Tff7b72-Tff7b72> Tb4b4b4{
Te6edf3debugLineBufTb4b4b4.Te6edf3appendTb4b4b4(Te6edf3cTb4b4b4)
Tff7b72if Tb4b4b4(Te6edf3debugLineBufTb4b4b4.Te6edf3length Tff7b72>Tff7b72= Te6edf3MAX_DEBUG_LINE_LENGTHTb4b4b4) Tb4b4b4{
Te6edf3LoggerTb4b4b4.Te6edf3d Tb4b4b4{ Ta5d6ff"Tffd700$Te6edf3logTagTa5d6ff DeviceLog: Tffd700$Te6edf3debugLineBufTa5d6ff" Tb4b4b4}
Te6edf3debugLineBufTb4b4b4.Te6edf3clearTb4b4b4(Tb4b4b4)
Tb4b4b4}
Tb4b4b4}
Tb4b4b4}
Tb4b4b4}
Tb4b4b4}

Served by rngit 1.5.0 - Generated in 0.12s